package com.bw.ads;

import com.bw.bean.Mn5TM4Wide4;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.Collections;

public class AsyncIOFunc4_4 extends RichAsyncFunction<Mn5TM4Wide4, Mn5TM4Wide4> {
    Connection conn = null;
    PreparedStatement ps1 = null;
    PreparedStatement ps2 = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName("com.mysql.jdbc.Driver");
        conn = DriverManager.getConnection("jdbc:mysql://hadoop102:3306/tms01?characterEncoding=utf8", "root", "123456");
        ps1 = conn.prepareStatement("select region_id from base_organ where id = ?");
        ps2 = conn.prepareStatement("select name from base_region_info where id = ?");
    }

    @Override
    public void close() throws Exception {
        ps1.close();
        conn.close();
    }

    @Override
    public void asyncInvoke(Mn5TM4Wide4 mn5TM4Wide, ResultFuture<Mn5TM4Wide4> resultFuture) throws Exception {
        ps1.setLong(1,mn5TM4Wide.getOrg_id());
        ResultSet rs1 = ps1.executeQuery();
        if(rs1.next()){
            Long region_id = rs1.getLong("region_id");
            Long area_code = Long.valueOf(region_id.toString().substring(0, 4) + "00");
            ps2.setLong(1,area_code);
            ResultSet rs2 = ps2.executeQuery();
            if(rs2.next()){
                String name = rs2.getString("name");
                if(name.equals("市辖区")){
                    Long area_code2 = Long.valueOf(region_id.toString().substring(0, 2) + "0000");
                    ps2.setLong(1,area_code2);
                    ResultSet rs3 = ps2.executeQuery();
                    if(rs3.next()){
                        name = rs3.getString("name");
                    }
                }
                mn5TM4Wide.setCity_name(name);
            }
        }
        resultFuture.complete(Collections.singleton(mn5TM4Wide));
    }
}
